/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.analysys.presto.connector.hbase.adaptive;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

import static com.analysys.presto.connector.hbase.adaptive.HColumn.ROWKEY_COLUMNNAME;

public class HTableInfo
        implements Serializable
{
    private static final long serialVersionUID = 20210412230798L;
    private String tableName;
    private String schemaName;
    private String rowKeyFormat;
    private String rowKeyFirstCharRange = "a~z,0~9";
    private List<HColumn> hColumnList;

    private String indexColumnName;
    private int indexColumnLength;
    private String pkColumnName;
    private int pkColumnLength;

    public HTableInfo(Builder builder)
    {
        this.tableName = builder.tableName;
        this.schemaName = builder.schemaName;
        this.rowKeyFormat = builder.rowKeyFormat;
        this.rowKeyFirstCharRange = builder.rowKeyFirstCharRange;
        this.hColumnList = builder.hColumnList;

        separateRowkeyFormat(this.rowKeyFormat);
    }

    private void separateRowkeyFormat(String rowKeyFormat)
    {
        if (this.rowKeyFormat == null) {
            this.rowKeyFormat = "";
        }
        else {
            this.indexColumnName = (rowKeyFormat.split("\\|")[1]).split("-")[0];
            this.indexColumnLength = Integer.parseInt((rowKeyFormat.split("\\|")[1]).split("-")[1]);
            this.pkColumnName = (rowKeyFormat.split("\\|")[2]).split("-")[0];
            this.pkColumnLength = Integer.parseInt((rowKeyFormat.split("\\|")[2]).split("-")[1]);
        }
    }

    public HTableInfo(String schemaName, String tableName)
    {
        this.tableName = tableName;
        this.schemaName = schemaName;
    }

    public HTableInfo(String tableName)
    {
        this.tableName = tableName;
        this.schemaName = "default";
    }

    public List<HColumn> gethColumnList()
    {
        return hColumnList;
    }

    public String getRowKeyColName()
    {
        return ROWKEY_COLUMNNAME;
    }

    public boolean equals(String schemaName, String tableName)
    {
        return this.schemaName.equals(schemaName) && this.tableName.equals(tableName);
    }

    public String getTableName()
    {
        return tableName;
    }

    public String getSchemaName()
    {
        return schemaName;
    }

    public String getRowKeyFormat()
    {
        return rowKeyFormat;
    }

    public String getRowKeyFirstCharRange()
    {
        return rowKeyFirstCharRange;
    }

    public String getIndexColumnName()
    {
        return indexColumnName;
    }

    public int getIndexColumnLength()
    {
        return indexColumnLength;
    }

    public String getPkColumnName()
    {
        return pkColumnName;
    }

    public int getPkColumnLength()
    {
        return pkColumnLength;
    }

    @Override
    public boolean equals(Object o)
    {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        HTableInfo hTableInfo = (HTableInfo) o;
        return Objects.equals(tableName, hTableInfo.tableName) &&
                Objects.equals(schemaName, hTableInfo.schemaName);
    }

    @Override
    public String toString()
    {
        return "HTable{" +
                "tableName='" + tableName + '\'' +
                ", schemaName='" + schemaName + '\'' +
                ", rowKeyFormat='" + rowKeyFormat + '\'' +
                ", rowKeyFirstCharRange='" + rowKeyFirstCharRange + '\'' +
                ", hColumnList=" + hColumnList +
                '}';
    }

    /**
     * A Builder to create HTable.
     * Example:
     * HTable hTable = new HTable.Builder("table")
     * .setSchemaName("schema")          // optional, default: "default"
     * .setRowKeyFormat("xx")            // optional, default: ""
     * .setRowKeyFirstCharRange("xx")    // optional, default: "a~z,0~9"
     * .addRowKey("varchar")
     * .addColumn("cf","column1","varcahr")
     * .addColumn("cf","column2","long")
     * .build();
     */
    public static class Builder
    {
        private String tableName;
        private String schemaName = "default";
        private String rowKeyFormat = "";
        private String rowKeyFirstCharRange = "a~z,0~9";
        private List<HColumn> hColumnList = new ArrayList<>();

        public Builder(String tableName)
        {
            this.tableName = tableName;
        }

        public Builder setSchemaName(String schemaName)
        {
            this.schemaName = schemaName;
            return this;
        }

        public Builder setRowKeyFormat(String rowKeyFormat)
        {
            this.rowKeyFormat = rowKeyFormat;
            return this;
        }

        public Builder setRowKeyFirstCharRange(String rowKeyFirstCharRange)
        {
            this.rowKeyFirstCharRange = rowKeyFirstCharRange;
            return this;
        }

        public Builder addColumn(String family, String columnName, String type)
        {
            hColumnList.add(new HColumn(family, columnName, type, "false"));
            return this;
        }

        public Builder addRowKey(String type)
        {
            hColumnList.add(new HColumn("", ROWKEY_COLUMNNAME, type, "true"));
            return this;
        }

        public HTableInfo build()
        {
            return new HTableInfo(this);
        }
    }
}
